package com.example.dddsample.application.event;

import com.alibaba.fastjson.JSON;
import com.example.dddsample.gateway.Handler.KafkaSendResultHandler;
import lombok.SneakyThrows;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

/**
 * @author: bebm
 * @date: 2022/4/2 14:32
 * @description:
 */
@Component
public class CustomerKafkaProducerEvent {

    @Value("${spring.kafka-topic.consumer.customerOperTopic}")
    private String customerOperTopic;

    @Resource
    private KafkaTemplate<Object, Object> template;

    @Autowired
    private KafkaSendResultHandler kafkaSendResultHandler;

    /**
     * simple
     *  开启事务
     * @param object
     */
    //@Transactional
    @SneakyThrows
    public void customerOperProducerProperties(Object object) {
        SendResult<Object,Object> result=template.send(customerOperTopic, JSON.toJSONString(object)).get();
        RecordMetadata recordMetadata=result.getRecordMetadata();
        ProducerRecord producerRecord=result.getProducerRecord();
        throw new RuntimeException("error");
        //todo


    }

    /**
     * complex
     * 发送
     *
     * @param object
     */
    public void customerOperProducerComplex(Object object) {
        //设置监听方法
        template.setProducerListener(kafkaSendResultHandler);
        //String topic, Integer partition, Long timestamp, K key, V data
        //partition 指定分发到哪个分区
        //使用时间戳可以进行日志切分
        //使用key,相同的key会分发到相同分区
        template.send(customerOperTopic, 0, System.currentTimeMillis(), 0, JSON.toJSONString(object));
        //


    }


}
